package net.butfly.bus.server;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.nio.channels.ClosedByInterruptException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import net.butfly.bus.Request;
import net.butfly.bus.Response;
import net.butfly.bus.config.bean.TimeoutBean;
import net.butfly.bus.context.Constants;
import net.butfly.bus.context.Constants.InternalTX;
import net.butfly.bus.context.ContextManager;
import net.butfly.bus.exception.BusRuntimeException;
import net.butfly.bus.filter.FilterBase;
import net.butfly.bus.filter.FilterChain;
import net.butfly.bus.server.invoker.Invoker;
import net.butfly.bus.server.invoker.InvokerFactory;
import net.butfly.bus.util.LogUtils;
import net.butfly.bus.util.ObjectUtils;

import org.dom4j.DocumentException;

public class ServerInvoker {
	private static LogUtils logger = LogUtils.getLog();
	private static ServerInvoker instance = new ServerInvoker();
	private static ObjectUtils ofh = new ObjectUtils();
	private ContextManager contextManager = ContextManager.getInstance();
	private FilterChain filterChain;
	private ThreadLocal<ExecutorService> exePool = new ThreadLocal<ExecutorService>();
	private static final FilterBase filter = new ServerBase();

	public static ServerInvoker getInstance() {
		return instance;
	}

	public Response execute(Request request) {
		long now = System.currentTimeMillis();

		if (request == null) { throw new BusRuntimeException(Constants.UserError.BAD_REQUEST, "request can not be null"); }
		String flowNo = request.getCurrentFlowNo();
		logger.info(LogUtils.getInstance().getFlowNo(
				flowNo != null ? flowNo : Constants.getFlowNo(request.getTx(), request.getVersion()))
				+ "bus server started.");
		if (logger.isDebugEnabled()) {
			StringBuilder sb = new StringBuilder();
			sb.append("\n\tdetail of current invkoing: ").append("\n\t\ttx context: ")
					.append(request.getContext().toString()).append("\n\t\ttx code: ").append(request.getTx())
					.append("\n\t\ttx version: ").append(null == request.getVersion() ? "NULL" : request.getVersion())
					.append("\n\t\tparametres: ");
			if (request.getArgs() != null) {
				for (Object to : request.getArgs()) {
					if (null != to) {
						sb.append("\n\t\t\t").append(to);
					}
				}
			} else {
				sb.append("[empty]");
			}
			logger.debug(LogUtils.getInstance().getFlowNo() + sb.toString());
		}

		contextManager.loadContext(request.getContext());
		Response response = null;
		try {
			if (ServerConfig.isSimulate()) {
				response = simulate(request);
			} else {
				// Response response = doExecute(request);
				if (filterChain == null) filterChain = new FilterChain(ServerConfig.getFilterBeanList());
				// Response response = doExecuteInTask(request);
				response = doExecute(request);
			}
			if (response.getErrorCode() == null && ServerConfig.getDefaultErrorCode() != null) {
				response.setErrorCode(ServerConfig.getDefaultErrorCode());
			}
		} catch (Exception ex) {
			response = response == null ? new Response() : response;
			response.setExecuteFail();
			response.setErrorCode(Constants.SystemError.UNKNOW_CAUSE);
			response.setErrorMessage("EBOX运行错误.");
			if (ServerConfig.isRunModeOfDebug()) {
				response.setErrorStack(ex);
			}
			logger.error(LogUtils.getInstance().getFlowNo() + "EBOX运行错误.", ex);
		} finally {
			contextManager.clearContext();
			logger.info(LogUtils.getInstance().getFlowNo() + "EBox Server运行结束，耗时 " + (System.currentTimeMillis() - now)
					+ "毫秒.");
			LogUtils.getInstance().clearFlowNo();
		}
		return response;
	}

	private Response simulate(Request request) throws IllegalAccessException, InvocationTargetException,
			InstantiationException, ClassNotFoundException, IOException, DocumentException {
		Response response = new Response();
		String filename = request.getTx() + "-" + request.getVersion() + ".xml";
		response.setResult(ofh.parse(ServerConfig.getSimulateDir()
				+ (ServerConfig.getSimulateDir().endsWith("\\") ? "" : "\\") + filename));
		response.setExecuteSuccess();
		return response;
	}

	private Response doExecute(Request request) {
		Response response = new Response();
		try {
			filterChain.execute(request, response, filter);
			response.setExecuteSuccess();
		} catch (BusRuntimeException ex) {
			response.setExecuteFail();
			response.setErrorCode(ex.getCode());
			if (ServerConfig.isRunModeOfDebug()) {
				response.setErrorMessage("错误号:" + ex.getCode() + "\n错误信息:" + ex.getMessage() + "\n" + (ex));
				response.setErrorStack(ex);
			} else {
				response.setErrorMessage("错误号:" + ex.getCode() + "\n错误信息:" + ex.getMessage());
			}

			logger.error(LogUtils.getInstance().getFlowNo() + "运行错误：", ex);
		} catch (InterruptedException ex) {
			response.setExecuteFail();
			response.setErrorCode(Constants.SystemError.TIMEOUT_FOR_WAIT);
			response.setErrorMessage("EBOX请求运行超时。");
			if (ServerConfig.isRunModeOfDebug()) {
				response.setErrorStack(ex);
			}
			logger.error(LogUtils.getInstance().getFlowNo() + "运行超时！");
		} catch (ClosedByInterruptException ex) {
			response.setExecuteFail();
			response.setErrorCode(Constants.SystemError.TIMEOUT_FOR_IO);
			response.setErrorMessage("EBOX请求因网络或其他I/O操作超时）");
			if (ServerConfig.isRunModeOfDebug()) {
				response.setErrorStack(ex);
			}
			logger.error(LogUtils.getInstance().getFlowNo() + "I/O操作超时！");
		} catch (Exception ex) {
			response.setExecuteFail();
			response.setErrorCode(Constants.SystemError.UNKNOW_CAUSE);
			if (ServerConfig.isRunModeOfDebug()) {
				response.setErrorMessage(ex.getMessage() + "\n" + (ex));
				response.setErrorStack(ex);
			} else {
				response.setErrorMessage(ex.getMessage());
			}
			logger.error(LogUtils.getInstance().getFlowNo() + "运行错误：", ex);
		}
		return response;
	}

	// TODO: not invoked, clean?
	protected Response doExecuteInTask(Request request) {
		Response response = new Response();
		TimeoutBean tb = ServerConfig.getTimeoutBean(request.getTx());

		ExecutorService executorService = exePool.get();

		if (executorService == null) {
			executorService = Executors.newSingleThreadExecutor();
			exePool.set(executorService);
		}

		List<Callable<Response>> taskList = new ArrayList<Callable<Response>>();

		Task tw = new Task(request);
		taskList.add(tw);

		try {
			if (tb != null) {
				logger.info(LogUtils.getInstance().getFlowNo() + "设置超时时间:" + tb.getValue() + "毫秒.");
				List<Future<Response>> futures = executorService.invokeAll(taskList, tb.getValue(),
						TimeUnit.MILLISECONDS);
				Future<Response> ft = futures.get(0);
				response = ft.get();
				if (ft.isCancelled()) {
					logger.error(LogUtils.getInstance().getFlowNo() + "调用业务逻辑超时(" + tb.getValue() + "毫秒),强制返回.");
					response.setExecuteFail();
					response.setErrorCode(Constants.BusinessError.INVOKE_TIMEOUT);
					response.setErrorMessage("请求超时(" + tb.getValue() + "毫秒),强制返回.");
					executorService.shutdown();
					try {
						while (null != tw && tw.isAlive()) {
							tw.interrupt();
						}
						tw = null;
					} catch (SecurityException ex) {
						response.setExecuteFail();
						response.setErrorCode(Constants.SystemError.TIMEOUT_CAN_NOT_STOP);
						response.setErrorMessage("bus invoking timeout, but the thread can not be terminated." + "\n"
								+ (ex));
						response.setErrorStack(ex);
						logger.error(LogUtils.getInstance().getFlowNo() + " failure：", ex);
					}
				}
			} else {
				response = executorService.invokeAll(taskList).get(0).get();
				tw = null;
			}

		} catch (ExecutionException ie) {
			response.setExecuteFail();
			response.setErrorCode(Constants.SystemError.UNKNOW_CAUSE);
			response.setErrorMessage("bus thread invoking exception." + "\n" + (ie));
			response.setErrorStack(ie);
			logger.error(LogUtils.getInstance().getFlowNo() + " failure：", ie);
		} catch (InterruptedException ie) {
			response.setExecuteFail();
			response.setErrorCode(Constants.SystemError.UNKNOW_CAUSE);
			response.setErrorMessage("bus thread invoking exception." + "\n" + (ie));
			response.setErrorStack(ie);
			logger.error(LogUtils.getInstance().getFlowNo() + " failure：", ie);
		}
		return response;

	}

	private class Task extends Thread implements Callable<Response> {
		private Request request;

		private Task(Request request) {
			this.request = request;
		}

		public Response call() throws Exception {
			return doExecute(request);
		}
	}

	private static class ServerBase extends FilterBase {
		@Override
		public void execute(Request request, Response response) {
			String currentNodeId = request.getCurrentNodeId();
			if (null == currentNodeId) {
				String[] nodes = ServerConfig.getNodeIdsByTxMatching(request.getTx());
				if (null != nodes && nodes.length > 0) currentNodeId = nodes[0];
			}
			if (null == currentNodeId) try {
				invokeInternalTX(request.getTx(), request, response);
				return;
			} catch (IllegalArgumentException ex) {
				throw new BusRuntimeException(Constants.SystemError.ROUTE_NOT_FOUND, "could not route to a bus node.");
			}

			logger.info(LogUtils.getInstance().getFlowNo() + "current bus node: " + currentNodeId + ".");
			Invoker invoker = InvokerFactory.getInstance().getInvoker(currentNodeId);
			if (null != invoker) {
				long t0 = System.currentTimeMillis();

				invoker.invoke(request, response);

				long t1 = System.currentTimeMillis();
				logger.info(LogUtils.getInstance().getFlowNo() + "(tx: " + request.getTx() + "-" + request.getVersion()
						+ ") spent " + (t1 - t0) + " ms.");
				/*
				 * now does not support remote connection to another server.
				 */
				// } else {
				// // looking for other nodes (remote) to route
				// logger.info(LogUtils.getInstance().getFlowNo()
				// +
				// "could not route to a bus node, begin to foind other nodes.");
				// String nodeId =
				// RouterFactory.getInstance().getRouter().route(request.getTx(),
				// request.getVersion());
				// if (null == nodeId || nodeId.equals(currentNodeId))
				// throw new
				// BusRuntimeException(Constants.SystemError.ROUTE_NOT_FOUND,
				// "could not route to a bus node.");
				// logger.info(LogUtils.getInstance().getFlowNo() +
				// "route to bus node: " + nodeId + ".");
				// // avoid to route to self.
				// NodeBean nodeBean = ServerConfig.getNodeBean(nodeId);
				// Connection sender = nodeBean != null ?
				// ConnectionFactory.getInstance().getConnection(nodeBean) :
				// null;
				// if (sender == null) {
				// throw new
				// BusRuntimeException(Constants.UserError.NODE_NOT_FOUND,
				// "can not found connection for node with id \"" + nodeId +
				// "\"");
				// }
				// logger.info(LogUtils.getInstance().getFlowNo() +
				// "use connection \"" + sender
				// + "\" to access node with id \"" + nodeId + "\".");
				// request.setCurrentNodeId(nodeId);
				// response.copyFrom(sender.send(request));
			}
		}

		private Response invokeInternalTX(String tx, Request request, Response response) {
			switch (InternalTX.parse(tx)) {
			case PING:
				response.setResult(true);
				break;
			case STAT:
				// should return status of the server.
				response.setResult("STAT");
				break;
			default:
				response.setResult(null);
				break;
			}
			logger.info(LogUtils.getInstance().getFlowNo() + "(internal tx: " + request.getTx() + "-"
					+ request.getVersion() + " finished.");
			return response;
		}
	}
}